package com.leyantech.ecp.agenx.translator.di;

import com.leyantech.ai.DialogueReply;
import com.leyantech.chaos.kafka.KafkaConsumerClient;
import com.leyantech.chaos.kafka.KafkaConsumerStringClient;
import com.leyantech.chaos.kafka.KafkaDialogueProducerClient;
import com.leyantech.ecp.agenx.common.converter.input.DialogueQuestionConverter;
import com.leyantech.ecp.agenx.common.converter.output.DialogueReplyConverter;
import com.leyantech.ecp.agenx.common.converter.output.PinDuoDuoReplyConverter;
import com.leyantech.ecp.agenx.common.di.HelaModule;
import com.leyantech.ecp.agenx.common.di.MessageDispatcherModule;
import com.leyantech.ecp.agenx.common.di.RedissonModule;
import com.leyantech.ecp.agenx.common.handler.ActionHandler;
import com.leyantech.ecp.agenx.common.message.EnumWsFnType;
import com.leyantech.ecp.agenx.common.message.MessageHelper;
import com.leyantech.ecp.agenx.common.serialization.DialogueReplyDeserializer;
import com.leyantech.ecp.agenx.session.client.SessionClientModule;
import com.leyantech.ecp.agenx.translator.consumer.ClientMessageConsumer;
import com.leyantech.ecp.agenx.translator.consumer.DialogueMessageConsumer;
import com.leyantech.ecp.agenx.translator.converter.input.PinDuoDuoQuestionConverter;
import com.leyantech.ecp.agenx.translator.filter.client.ClientMessageFilter;
import com.leyantech.ecp.agenx.translator.filter.client.DuplicateMessageFilter;
import com.leyantech.ecp.agenx.translator.filter.client.ExpireMessageFilter;
import com.leyantech.ecp.agenx.translator.filter.client.TransferMessageFilter;
import com.leyantech.ecp.agenx.translator.handler.BuyerMessageHandler;

import com.ctrip.framework.apollo.Config;
import com.ctrip.framework.apollo.ConfigService;
import com.google.inject.AbstractModule;
import com.google.inject.Inject;
import com.google.inject.Injector;
import com.google.inject.Provider;
import com.google.inject.Provides;
import com.google.inject.Singleton;
import com.google.inject.TypeLiteral;
import io.vertx.core.json.JsonObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.MDC;

import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-15.
 */
public class TranslatorModule extends AbstractModule {

  @Override
  protected void configure() {

    install(new HelaModule());

    install(new SessionClientModule());

    install(new MessageDispatcherModule());

    install(new RedissonModule());

    final KafkaDialogueProducerClient dialogueProducerClient = KafkaDialogueProducerClient
        .newBuilder("dialogue")
        .build();
    bind(KafkaDialogueProducerClient.class).toInstance(dialogueProducerClient);

    bind(DialogueReplyConverter.class)
        .to(PinDuoDuoReplyConverter.class)
        .asEagerSingleton();

    bind(DialogueQuestionConverter.class)
        .to(PinDuoDuoQuestionConverter.class)
        .asEagerSingleton();

    bind(new TypeLiteral<KafkaConsumerClient<String, DialogueReply>>() {
    })
        .toProvider(dialogueReplyConsumerProvider())
        .asEagerSingleton();

    bind(KafkaConsumerStringClient.class)
        .toProvider(clientMessageConsumerProvider())
        .asEagerSingleton();

  }

  public Provider<KafkaConsumerClient<String, DialogueReply>> dialogueReplyConsumerProvider() {

    return new Provider<KafkaConsumerClient<String, DialogueReply>>() {

      @Inject
      DialogueMessageConsumer processor;

      @Override
      public KafkaConsumerClient<String, DialogueReply> get() {
        return KafkaConsumerClient.<String, DialogueReply>newBuilder("dialogue")
            .setKeyDeserializer(StringDeserializer.class)
            .setValueDeserializer(DialogueReplyDeserializer.class)
            .setProcessor(record -> {
              try {
                final DialogueReply reply = record.value();
                MDC.put("sid", reply.getStoreId());
                MDC.put("aid", reply.getAssistantId());
                processor.accept(record);
              } finally {
                MDC.clear();
              }
            })
            .build();
      }
    };
  }

  public Provider<KafkaConsumerStringClient> clientMessageConsumerProvider() {

    return new Provider<KafkaConsumerStringClient>() {

      List<ConsumerRecord<String, String>> warmupRecords() {
        final Config config = ConfigService.getConfig("warmup");

        final String warmupBody = config.getProperty("translator.question.body", "{}");

        final JsonObject message = MessageHelper.buildMessage(
            EnumWsFnType.NORMAL_BUYER_MSG,
            new JsonObject(warmupBody)
        );

        final ConsumerRecord<String, String> record = new ConsumerRecord<>(
            "",
            0,
            0,
            config.getProperty("translator.question.key", "191781350.100817226"),
            message.toString()
        );
        return Collections.singletonList(record);
      }

      @Inject
      ClientMessageConsumer processor;

      @Override
      public KafkaConsumerStringClient get() {
        return KafkaConsumerStringClient.newBuilder("translator")
            .setProcessor(record -> {
              try {
                final String key = record.key();
                // {店铺号}.{客服号}
                MDC.put("cid", key);
                processor.accept(record);
              } finally {
                MDC.clear();
              }
            })
            .setWarmupRecord(warmupRecords())
            .build();
      }
    };
  }

  @Singleton
  @Provides
  public List<ClientMessageFilter> provideClientMessageFilters(Injector injector) {
    return Stream.of(
            ExpireMessageFilter.class,
            DuplicateMessageFilter.class,
            TransferMessageFilter.class
        )
        .map(injector::getInstance)
        .collect(Collectors.toList());
  }

  @Singleton
  @Provides
  public List<ActionHandler> provideActionHandlers(Injector injector) {
    return Stream.of(BuyerMessageHandler.class)
        .map(injector::getInstance)
        .collect(Collectors.toList());
  }

}
